package thread;

import thread.biz.DOrder;
import thread.biz.Diff;
import thread.biz.POrder;

import java.util.concurrent.*;

/**
 * semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源.这个跟队列有点像
 */
public class CheckBatch3 extends AbstractCheckBatch {
    @Override
    public void doProcess() throws Exception {
        System.out.println("---- semaphore(信号量)处理 ----");
        ExecutorService service = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(10);
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (int id : ids) {
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    try {
                        semaphore.acquire();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    check(id, executor);
                    semaphore.release();
                }
            };
            service.execute(runnable);//放在线程池中去维护
        }
        service.shutdown();
        //加上了这行，main线程就会阻塞，后面计算执行时间的代码无法执行。直到所有线程执行结束或者到了1分钟，才能计算时间。从而达到目的
        service.awaitTermination(1, TimeUnit.MINUTES);
        executor.shutdown();
    }

    @Override
    protected void doFinal() {
        // NoOP
    }

    public Integer check(Integer id, ExecutorService executor) {
        CompletableFuture<POrder> f1 = CompletableFuture.supplyAsync(() -> pOrderService.getPOrder(id), executor);
        CompletableFuture<DOrder> f2 = CompletableFuture.supplyAsync(() -> dOrderService.getDOrder(id), executor);
        try {
            POrder pOrder = f1.get();
            DOrder dOrder = f2.get();
            Diff diff = check(pOrder, dOrder);
            if (diff != null)
                // 2ms
                save(diff);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return 1;
    }
}
